package a_o2odata_deal.utils

import a_aa_amainpackage.a_o2odata_deal.config.config.source_result
import org.apache.spark.sql.{DataFrame, SQLContext}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/12/3 08:16
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object handle_zhibo {

  def pricedeal(priceText: String): String = {
    var price = "-1"
    if (!priceText.equals("-1")) {
      if (priceText.contains("-")) {
        price = priceText.split("-")(0)
      } else {
        price = priceText
      }
    } else {
      price = "-1"
    }
    price
  }

  def handle_zhibo(sqlContext:SQLContext):DataFrame={

    sqlContext.udf.register("pricedeal", pricedeal _)
    //读取关联完品牌的数据
    val brand_data = sqlContext.read.orc(s"${source_result}")
      .where("cast(sellCount as bigint)>0 and cast(salesAmount as double)>0 and cast(priceText as double)>0")
      .dropDuplicates("good_id")

    //读取预售数据
    /*val yushou_data = sqlContext.read.json("s3a://o2o-dataproces-group/panzonghao/last_yushou_month11_result")*/

    brand_data.createOrReplaceTempView("brand_data")
    //yushou_data.createOrReplaceTempView("yushou_data")

    //与预售关联上的进行处理
   /* val data1 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.good_id is null then cast(t1.salesAmount as double)
         |else (cast(t1.sellCount as bigint)-cast(t2.sellCount as bigint))*cast(t1.priceText as double)+cast(t2.salesAmount as double)
         |end as sales
         |from brand_data t1
         |left join
         |yushou_data t2
         |on t1.good_id=t2.good_id
       """.stripMargin).where("sales>0").drop("salesAmount")
      .withColumnRenamed("sales","salesAmount")

    val data2 = sqlContext.sql(
      s"""
         |select
         |t1.*
         |from yushou_data t1
         |left join
         |brand_data t2
         |on t1.good_id=t2.good_id
         |where t2.good_id is null
       """.stripMargin)*/

    //加载新品标签
    val newProduct_data = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.is_newProduct_path}")

    /*data1.createOrReplaceTempView("data1")
    data2.createOrReplaceTempView("data2")*/
    newProduct_data.createOrReplaceTempView("newProduct_data")

    val result_data1 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.good_id is null then false
         |else t2.is_newProduct
         |end as is_newProducts
         |from brand_data t1
         |left join
         |newProduct_data t2
         |on t1.good_id=t2.good_id
       """.stripMargin).drop("is_newProduct","is_showLive").withColumnRenamed("is_newProducts","is_newProduct")

    result_data1.createOrReplaceTempView("result_data1")


    //关联直播
    val data_detail = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_detail_path}")
    val data_id = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_id_path}")

    data_id.createOrReplaceTempView("data_id")
    data_detail.createOrReplaceTempView("data_detail")
    //首先统计出单品在多少个直播间出现过

    val mid_data = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |pricedeal(t2.livePriceText) as live_priceText,
         |t2.good_id as good_ids
         |from data_id t1
         |inner join
         |data_detail t2
         |on t1.liveId=t2.liveId
       """.stripMargin).drop("good_id").withColumnRenamed("good_ids","good_id").where("good_id!='-1'")

    mid_data.createOrReplaceTempView("mid_data")

    sqlContext.sql(
      s"""
         |select
         |*,
         |--直播次数
         |sum(cast(liveCount as bigint)) over(partition by good_id) as liveCounts,
         |--观看人数
         |sum(cast(viewerCount as bigint)) over(partition by good_id) as viewcount,
         |--直播均价
         |avg(cast(live_priceText as double)) over(partition by good_id) as liveAvgPriceText
         |from mid_data
       """.stripMargin).drop("liveCount").withColumnRenamed("liveCounts","liveCount").dropDuplicates("good_id").createOrReplaceTempView("value")

    sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.userId is null then false
         |else true
         |end as is_selfLive
         |from value t1
         |left join
         |result_data1 t2
         |on t1.anchorId=t2.userId
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("data")
    val result = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |cast(t1.sellCount as bigint) as sellCounts,
         |cast(t2.live_priceText as double) as live_priceText,
         |cast(t2.liveCount as bigint) as liveCount,
         |cast(t2.viewcount as bigint) as viewcount,
         |cast(t2.liveAvgPriceText as double) as liveAvgPriceText,
         |cast(t2.anchorFans as bigint) as anchorFans,
         |t2.is_selfLive,
         |t2.roomType,
         |t2.liveId,
         |t2.anchorId,
         |t2.liveTime,
         |t2.roomTypeName,
         |t2.nick,
         |t2.liveTitle,
         |t2.commodityCount,
         |case
         |when t2.good_id is null then 'false'
         |else 'true'
         |end as is_showLive
         |from result_data1 t1
         |left join
         |data t2
         |on t1.good_id=t2.good_id
       """.stripMargin).dropDuplicates("good_id").drop("sellCount")
      .withColumnRenamed("sellCounts","sellCount")

    /*val result_data2 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.good_id is null then false
         |else t2.is_newProduct
         |end as is_newProducts
         |from data2 t1
         |left join
         |newProduct_data t2
         |on t1.good_id=t2.good_id
       """.stripMargin).drop("is_newProduct","is_showLive","roomType","liveId","anchorId","liveTime",
      "roomTypeName","nick","liveTitle","commodityCount","promotion_price","viewerCount")
      .withColumnRenamed("is_newProducts","is_newProduct")

    result_data2.createOrReplaceTempView("result_data2")*/

    //关联直播
   /* val data4 = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_detail_path}")
    val data5 = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_id_path}")

    data4.createOrReplaceTempView("data4")
    data5.createOrReplaceTempView("data5")

    val data3 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.good_id as good_ids
         |from data5 t1
         |inner join
         |data4 t2
         |on t1.liveId=t2.liveId
       """.stripMargin).drop("good_id").withColumnRenamed("good_ids","good_id").where("good_id!='-1'")

    data3.createOrReplaceTempView("data3")

    import org.apache.spark.sql.functions._
    val result1 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |cast(t1.sellCount as bigint) as sellCounts,
         |cast(t2.viewerCount as bigint) as viewcount,
         |t2.roomType,
         |t2.liveId,
         |t2.anchorId,
         |t2.liveTime,
         |t2.roomTypeName,
         |t2.nick,
         |t2.liveTitle,
         |t2.commodityCount,
         |cast(t2.liveCount as bigint) as liveCount,
         |cast(t2.live_priceText as double) as live_priceText,
         |cast(t2.anchorFans as bigint) as anchorFans,
         |case
         |when t2.good_id is null then 'false'
         |else 'true'
         |end as is_showLive
         |from result_data1 t1
         |left join
         |data3 t2
         |on t1.good_id=t2.good_id
       """.stripMargin).dropDuplicates("good_id").drop("timeStamp","sellCount").withColumn("timeStamp",lit("1575043200"))
      .withColumnRenamed("sellCounts","sellCount")

    val result2 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |cast(t1.sellCount as bigint) as sellCounts,
         |cast(t2.viewerCount as bigint) as viewcount,
         |t2.roomType,
         |t2.liveId,
         |t2.anchorId,
         |t2.liveTime,
         |t2.roomTypeName,
         |t2.nick,
         |t2.liveTitle,
         |t2.commodityCount,
         |cast(t2.liveCount as bigint) as liveCount,
         |case
         |when t2.good_id is null then 'false'
         |else 'true'
         |end as is_showLive
         |from result_data2 t1
         |left join
         |data3 t2
         |on t1.good_id=t2.good_id
       """.stripMargin).dropDuplicates("good_id").drop("timeStamp","sellCount").withColumn("timeStamp",lit("1575043200"))
      .withColumnRenamed("sellCounts","sellCount")*/


    result
  }

}
